package com.atguigu.flink.chapter08;




import com.atguigu.flink.bean.HotItem;
import com.atguigu.flink.bean.UserBehavior;
import com.atguigu.flink.util.KKutil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import java.util.TreeSet;

public class Flink05_High_Project_Product_TopN_1 {

       public static void main(String[] args) {
               Configuration configuration = new Configuration();
               //web  UI端口
               configuration.setInteger("rest.prot",10000);
               StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
               env.setParallelism(1);

               env.readTextFile("input/UserBehavior.csv")
                       .map(line->{
                           String[] datas = line.split(",");
                           return new UserBehavior(
                                   Long.valueOf(datas[0]),
                                   Long.valueOf(datas[1]),
                                   Integer.valueOf(datas[2]),
                                   datas[3],
                                   Long.parseLong(datas[4])*1000
                           );
                       })
                       // 添加水印
                       .assignTimestampsAndWatermarks(
                               WatermarkStrategy
                                       .<UserBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                       //添加时间戳
                               .withTimestampAssigner((ub,ts)->ub.getTimestamp())
                       )
                       // 过滤行为，只要 pv 行为
                       .filter(ub ->"pv".equals(ub.getBehavior()))
                       //按照商品id进行分组，统计每个商品的点击量
                       .keyBy(UserBehavior::getItemId)
                       .window(SlidingEventTimeWindows.of(Time.hours(1),Time.minutes(5)))
                       .aggregate(new AggregateFunction<UserBehavior, Long, Long>() {
                                      @Override
                                      public Long createAccumulator() {
                                          return 0L;
                                      }

                                      @Override
                                      public Long add(UserBehavior value, Long acc) {
                                          return acc + 1;
                                      }

                                      @Override
                                      public Long getResult(Long acc) {
                                          return acc;
                                      }

                                      @Override
                                      public Long merge(Long a, Long b) {
                                          return null;
                                      }
                                  }, new ProcessWindowFunction<Long, HotItem, Long, TimeWindow>() {
                                      @Override
                                      public void process(Long key,
                                                          Context ctx,
                                                          Iterable<Long> elements,
                                                          Collector<HotItem> out) throws Exception {
                                          Long count = elements.iterator().next();
                                          long end = ctx.window().getEnd();
                                          out.collect(new HotItem(key,end,count));

                                      }
                                  }
                       )
                       //商品id太多了，商品在每个窗口排序
                       .keyBy(HotItem::getWEnd)
                       .process(new KeyedProcessFunction<Long, HotItem, String>() {


                           private ValueState<TreeSet<HotItem>> treeSetState;

                           //使用Treeset状态 存储所有值,可以自动排序的集合  set 自动去重因此，需要进行修改
                           @Override
                           public void open(Configuration parameters) throws Exception {
                               treeSetState = getRuntimeContext().getState(new ValueStateDescriptor<TreeSet<HotItem>>(
                                       "treeSetState", TypeInformation.of(new TypeHint<TreeSet<HotItem>>() {
                               })));
                           }

                           @Override
                           public void processElement(HotItem hotItem,
                                                      Context ctx,
                                                      Collector<String> out) throws Exception {
                               //第一个来了之后，应该注册一个定时器 （触发之后再进行排序）
                               TreeSet<HotItem> set = treeSetState.value();
                               if (set == null) {
                                   set = new TreeSet<HotItem>(new Comparator<HotItem>() {
                                       @Override
                                       public int compare(HotItem o1, HotItem o2) {
                                           return o1.getCount().equals(o2.getCount()) ? 1:o2.getCount().compareTo(o1.getCount());
                                       }
                                   });
                                   treeSetState.update(set);
                                   //注册定时器
                                   ctx.timerService().registerEventTimeTimer(hotItem.getWEnd() + 2000);
                               }
                               set.add(hotItem);

                               //对状态内的元素  做优化， 排序后只保留 前三
                               // 当set 集合长度等于4 ，则删除最后一个
                               if (set.size()==4) {
                                  // set.remove(set.size()-1);
                                   set.pollLast();
                               }
                           }

                           //定时器触发后，进行排序
                           @Override
                           public void onTimer(long timestamp,
                                               OnTimerContext ctx,
                                               Collector<String> out) throws Exception {

                               TreeSet<HotItem> set = treeSetState.value();
                               StringBuilder msg = new StringBuilder("---------\n");

                              // StringBuilder msg = new StringBuilder("------------\n");

                               for (HotItem hotItem : set) {
                                   msg.append(hotItem).append("\n");
                               }
                               out.collect(msg.toString());
                           }
                       })
                       .print();





               try {
                   env.execute();
               } catch (Exception e) {
                   e.printStackTrace();
               }


           }


}
